package com.Parser;

import com.Pojo.FileBeat;
import com.Pojo.IISLog;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;

import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;


/**
 * Filebeat--IIS的module默认配置文件
 * type: log
 * paths:
 * {{ range $i, $path := .paths }}
 *  - {{$path}}
 * {{ end }}
 * exclude_files: [".gz$"]
 * exclude_lines: ["^#"]
 * processors:
 *   - add_fields:
 *       target: ''
 *       fields:
 *         ecs.version: 1.12.0
 *
 * 说明：
 * exclude_lines：在输入中排除符合正则表达式列表的那些行。
 * */

public class FilebeatJSONKeyValueIISDeserializationSchema implements KafkaDeserializationSchema<String> {

    private static final long serialVersionUID = 1509391548173891945L;




    @Override
    public boolean isEndOfStream(String s) {
        return false;
    }

    @Override
    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        //Flink解析器固定返回String类型 提醒：接口是泛型，具体返回类型可自己定义
        String node = "" ;

        if (record.value() != null) {
            //使用fastjson解析日志处理，获取实际的Filebeat日志然后转为json字符串
            FileBeat fileBeat = JSON.parseObject(record.value(), FileBeat.class);
            IISLog log = new IISLog();
            String [] message_1 = fileBeat.getMessage().split(" ");
            LocalDateTime dateTime = LocalDateTime.parse(message_1[0]+" "+message_1[1],
                    DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            //IIS默认是W3C日志文件格式，默认为UTC，比中国少8H。需要手动加回8小时再插入数据库
            LocalDateTime new_time = dateTime.plusHours(8);
            log.setDatetime(new_time);
            log.setS_ip(message_1[2]);
            log.setCs_method(message_1[3]);
            log.setCs_uri_stem(message_1[4]);
            log.setCs_uri_query(message_1[5]);
            log.setS_port(message_1[6]);
            log.setCs_username(message_1[7]);
            log.setC_ip(message_1[8]);
            log.setUser_Agent(message_1[9]);
            log.setReferer(message_1[10]);
            log.setStatus(message_1[11]);
            log.setTime_taken(message_1[14]);

            node = JSONObject.toJSONString(log);
            }

        return node;
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return TypeExtractor.getForClass(String.class); //显式确定返回的数据类型
    }
}
